package com.dcloud.streamweb.taskmanage.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.dcloud.streamweb.core.jobserver.JobServerAO;
import com.dcloud.streamweb.core.jobserver.JobServerAOFactory;
import com.dcloud.streamweb.enums.DeployModeEnum;
import com.dcloud.streamweb.enums.JobConfigStatus;
import com.dcloud.streamweb.enums.SysConfigEnum;
import com.dcloud.streamweb.enums.SysErrorEnum;
import com.dcloud.streamweb.exception.BizException;
import com.dcloud.streamweb.sysconfig.service.ISystemConfigService;
import com.dcloud.streamweb.sysconfig.service.impl.SystemConfigServiceImpl;
import com.dcloud.streamweb.taskmanage.entity.JobConfig;
import com.dcloud.streamweb.taskmanage.entity.SavepointBackup;
import com.dcloud.streamweb.taskmanage.entity.dto.JobConfigDTO;
import com.dcloud.streamweb.taskmanage.entity.req.StartJobConfigReq;
import com.dcloud.streamweb.taskmanage.entity.req.StopJobConfigReq;
import com.dcloud.streamweb.taskmanage.mapper.JobConfigMapper;
import com.dcloud.streamweb.taskmanage.mapper.SavepointBackupMapper;
import com.dcloud.streamweb.taskmanage.service.IJobConfigService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.jeecg.common.system.vo.LoginUser;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springframework.transaction.annotation.Transactional;

import java.util.Collections;
import java.util.List;

/**
 * @Description: job_config
 * @Author: jeecg-boot
 * @Date:   2024-06-26
 * @Version: V1.0
 */
@Slf4j
@Service
public class JobConfigServiceImpl extends ServiceImpl<JobConfigMapper, JobConfig> implements IJobConfigService {

    @Autowired
    JobConfigMapper jobConfigMapper;

    @Autowired
    SavepointBackupMapper savepointBackupMapper;

    @Autowired
    ISystemConfigService systemConfigService;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void addJob(JobConfigDTO jobConfigDTO) {

        Long jobConfigId = this.addJobConfig(jobConfigDTO);
//        jobAlarmConfigService
//                .upSertBatchJobAlarmConfig(jobConfigDTO.getAlarmTypeEnumList(), jobConfigId);

    }

    @Override
    public void startJob(StartJobConfigReq startJobConfigReq, LoginUser sysUser) {

        //根据id获取任务详情
        JobConfigDTO jobConfigDTO = this.getJobConfigById(startJobConfigReq.getId());

        //根据运行模型，从jobserver工厂中获取接口
        JobServerAO jobServerAO = JobServerAOFactory.getJobServerAO(jobConfigDTO.getDeployModeEnum());

        jobServerAO.start(startJobConfigReq.getId(), startJobConfigReq.getSavepointId(), sysUser.getUsername());

    }

    @Override
    public JobConfigDTO getJobConfigById(Long id) {
        if (id == null) {
            throw new BizException(SysErrorEnum.JOB_CONFIG_PARAM_IS_NULL);
        }

        JobConfigDTO jobConfigDTO = JobConfigDTO.toDTO(jobConfigMapper.selectByPrimaryKey(id));
        if (jobConfigDTO == null) {
            throw new BizException(SysErrorEnum.JOB_CONFIG_JOB_IS_NOT_EXIST);
        }
        return jobConfigDTO;
    }

    @Override
    public void updateStatusByStart(Long id, String userName, Long jobRunLogId, Integer version) {
        int num = jobConfigMapper.updateStatusByStart(id, JobConfigStatus.STARTING.getCode(), userName,
                jobRunLogId, version);
        if (num < 1) {
            throw new BizException("启动状态更新失败");
        }

    }

    @Override
    public void updateJobConfigById(JobConfigDTO jobConfigDTO) {
        if (jobConfigDTO == null || jobConfigDTO.getId() == null) {
            throw new BizException(SysErrorEnum.JOB_CONFIG_PARAM_IS_NULL);
        }
        //任务id,查询任务详情
        JobConfig jobConfig = jobConfigMapper.selectByPrimaryKey(jobConfigDTO.getId());
        //判断原数据是否存在
        if (jobConfig == null) {
            throw new BizException(SysErrorEnum.JOB_CONFIG_JOB_IS_NOT_EXIST);
        }

        //根据新数据，或者原数据的运行模式 检查系统配置
        if (jobConfigDTO.getDeployModeEnum() == null) {
            this.checkSystemConfig(DeployModeEnum.valueOf(jobConfig.getDeployMode()));
        } else {
            this.checkSystemConfig(jobConfigDTO.getDeployModeEnum());
        }

        if (StringUtils.isNotEmpty(jobConfigDTO.getJobName())) {
            this.checkJobName(jobConfigDTO.getJobName(), jobConfigDTO.getId());
        }
        JobConfig jobConfigUpdate = JobConfigDTO.toEntity(jobConfigDTO);

        //修改数据
        jobConfigMapper.updateById(jobConfigUpdate);
    }

    @Override
    public void stop(StopJobConfigReq stopJobConfigReq, LoginUser sysUser) {
        //根据id获取任务详情
        JobConfigDTO jobConfigDTO = this.getJobConfigById(stopJobConfigReq.getId());

        //根据运行模型，从jobserver工厂中获取接口
        JobServerAO jobServerAO = JobServerAOFactory.getJobServerAO(jobConfigDTO.getDeployModeEnum());
        jobServerAO.stop(stopJobConfigReq.getId(), sysUser.getUsername());
    }

    @Override
    public void savepoint(Long id) {
        //根据id获取任务详情id
        JobConfigDTO jobConfigDTO = this.getJobConfigById(id);

        //根据运行模型，从jobserver工厂中获取接口
        JobServerAO jobServerAO = JobServerAOFactory.getJobServerAO(jobConfigDTO.getDeployModeEnum());
        jobServerAO.savepoint(id);
    }

    @Override
    public List<SavepointBackup> savepointHitory(Long jobConfigId) {
        List<SavepointBackup> savepointBackups = savepointBackupMapper.selectList(new LambdaQueryWrapper<SavepointBackup>()
                .eq(SavepointBackup::getJobConfigId, jobConfigId).orderByDesc(SavepointBackup::getBackupTime).last(" limit 5"));
        return savepointBackups;
    }

    public Long addJobConfig(JobConfigDTO jobConfigDTO) {
        if (jobConfigDTO == null) {
            return null;
        }
        this.checkJobName(jobConfigDTO.getJobName(), jobConfigDTO.getId());
        this.checkSystemConfig(jobConfigDTO.getDeployModeEnum());
        JobConfig jobConfig = JobConfigDTO.toEntity(jobConfigDTO);
        if (jobConfigDTO.getId() == null) {
            jobConfigMapper.insert(jobConfig);
        } else {
            jobConfigMapper.insertWithId(jobConfig);
        }
        this.insertJobConfigHistory(jobConfig.getId());
        return jobConfig.getId();
    }

    private void insertJobConfigHistory(Long id) {
        JobConfig jobConfig = jobConfigMapper.selectById(id);
        if (jobConfig == null) {
            log.warn("[insertJobConfigHistory] jobConfig is null id:{} ", id);
            return;
        }

        // jobConfigHistoryService.insertJobConfigHistory(JobConfigHistoryDTO.to(jobConfig));
    }


    /**
     * 检查任务名称是不是重复
     *
     * @author zhuhuipei
     * @date 2020-07-14
     * @time 13:56
     */
    private void checkJobName(String jobName, Long id) {
        long count = jobConfigMapper.selectCountByJobName(jobName, id);
        if (count >= 1) {
            throw new BizException(SysErrorEnum.JOB_CONFIG_JOB_NAME_IS_EXIST);
        }
    }

    /**
     * 检查配置文件
     *
     * @author zhuhuipei
     * @date 2020/11/4
     * @time 22:43
     */
    private void checkSystemConfig(DeployModeEnum deployModeEnum) {
        StringBuffer tips = new StringBuffer();
        String flinkHome = systemConfigService.getSystemConfigByKey(SysConfigEnum.FLINK_HOME.getKey());
        if (StringUtils.isEmpty(flinkHome)) {
            tips.append(" flinkHome、");
        }
        String webHome = systemConfigService.getSystemConfigByKey(
                SysConfigEnum.FLINK_STREAMING_PLATFORM_WEB_HOME.getKey());
        if (StringUtils.isEmpty(webHome)) {
            tips.append(" web应用安装的目录 ");
        }
        switch (deployModeEnum) {
            case YARN_PER:
            case YARN_APPLICATION:
                String yarnHttpAddress = systemConfigService.getSystemConfigByKey(
                        SysConfigEnum.YARN_RM_HTTP_ADDRESS.getKey());
                if (StringUtils.isEmpty(yarnHttpAddress)) {
                    tips.append(" yarn_rm_http_address url地址 ");
                }
                break;
            case LOCAL:
                String flinkHttpAddress = systemConfigService.getSystemConfigByKey(
                        SysConfigEnum.FLINK_REST_HTTP_ADDRESS.getKey());
                if (StringUtils.isEmpty(flinkHttpAddress)) {
                    tips.append(" flink_rest_http_address url地址 ");
                }
                break;
            case STANDALONE:
                String flinkHaHttpAddress =
                        systemConfigService
                                .getSystemConfigByKey(SysConfigEnum.FLINK_REST_HA_HTTP_ADDRESS.getKey());
                if (StringUtils.isEmpty(flinkHaHttpAddress)) {
                    tips.append(" flink_rest_ha_http_address url地址 ");
                }
                break;
            default:
                break;
        }
        if (StringUtils.isNotEmpty(tips.toString())) {
            throw new BizException("请在  系统管理-->系统设置 里面配置 " + tips.toString());
        }

    }


}
